import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, run}
import com.cw.realtime.common.bean.UserLoginBean
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.DateFormatUtil._
import com.cw.realtime.common.util.FlinkSinkUtil
import io.circe._
import io.circe.generic.auto._
import io.circe.parser._
import io.circe.syntax._
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration

object UserLogin {

  def main(args: Array[String]): Unit = {
    implicit val conf = RunnerConfig("dws_user_user_login_window", TOPIC_DWD_TRAFFIC_PAGE, 10024, ckp = true)
    run { (_, ds) =>
      // 清洗数据
      val jsonObjStream = etl(ds)

      // 设定水位线
      val jsonObjectStreamWM = withWatermark(jsonObjStream)

      // 判断独立用户和回流用户
      val loginStream = judgeUser(jsonObjectStreamWM)
//      loginStream.print()

      // 开窗聚合
      val aggStream = windowAggregate(loginStream)

      // sink
      aggStream.sinkTo(FlinkSinkUtil.getDorisSink(DWS_USER_USER_LOGIN_WINDOW))

    }



    def windowAggregate(ds: DataStream[UserLoginBean]) = {
      ds.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
        .reduce(_ + _, new AllWindowFunction[UserLoginBean, String, TimeWindow] {
          override def apply(window: TimeWindow, input: Iterable[UserLoginBean], out: Collector[String]): Unit = {
            val stt = window.getStart.toDateTime
            val edt = window.getEnd.toDateTime
            val curDate = System.currentTimeMillis.toDate
            input.foreach { e => out.collect(e.transWindowRange(stt, edt, curDate).asJson.noSpaces) }
          }
        })
    }


    def judgeUser(ds: DataStream[JsonObject]) = {
      ds.keyBy(_("common").get.asObject.get("mid").get.asString.get)
        .process(new KeyedProcessFunction[String, JsonObject, UserLoginBean] {
          type FuncType = KeyedProcessFunction[String, JsonObject, UserLoginBean]

          val stateDesc = new ValueStateDescriptor("login_date", classOf[String])
          var loginState = Option.empty[ValueState[String]]

          override def open(parameters: Configuration): Unit = {
            loginState = Some(getRuntimeContext.getState(stateDesc))
            // 因为要判断回流用户，所以状态要一直保存
          }

          override def processElement(value: JsonObject, ctx: FuncType#Context, out: Collector[UserLoginBean]): Unit = {
            for {
              state <- loginState
              common <- value("common")
              _ <- common.asObject
              ts <- value("ts")
              ts <- ts.asNumber
              ts <- ts.toLong
            } {
              val lastLogin = state.value
              val curLogin = ts.toDate
              val backCnt =
                if (lastLogin != null && ts - dateToTs(lastLogin) >= 7 * 24 * 60 * 60 * 1000) 1 else 0

              val uuCnt =
                if (lastLogin == null || curLogin != lastLogin) 1 else 0

              state.update(curLogin)
              if (backCnt | uuCnt == 1)
                out.collect(UserLoginBean(backCnt, uuCnt, ts))

            }

          }
        })
    }


    def withWatermark(ds: DataStream[JsonObject]) = {
      ds.assignTimestampsAndWatermarks(
        WatermarkStrategy.forBoundedOutOfOrderness(
          Duration.ofSeconds(3)).withTimestampAssigner(
          new SerializableTimestampAssigner[JsonObject] {
            override def extractTimestamp(element: JsonObject, recordTimestamp: Long): Long =
              element("ts").get.asNumber.get.toLong.get
          }))
    }


    def etl(ds: DataStream[String]) = {
      ds.flatMap { s =>
        for {
          json <- parse(s).toOption
          jsonObj <- json.asObject
          common <- jsonObj("common")
          commonObj <- common.asObject
          uid <- commonObj("uid")
          _ <- uid.asString
          ts <- jsonObj("ts")
          _ <- ts.asNumber
          page <- jsonObj("page")
          pageObj <- page.asObject
          lastPageId = pageObj("last_page_id") if
            lastPageId.isEmpty || lastPageId.get.isNull || lastPageId.get.asString.get == "login"
        } yield jsonObj
      }
    }


  }

}
